package exp1;

import java.io.IOException;
import java.util.ArrayList;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;


public class Lab3 {
	public static class Map extends Mapper<Object,Text,Text,Text>{

		public void map(Object key, Text value, Context context) throws IOException,InterruptedException{
			String []name = value.toString().split(" ");
			if(name[0].compareTo("child") != 0) {
				Text parent = new Text(name[1]);
				Text child = new Text(name[0]);
				context.write(parent, new Text("p+"+name[0]+"+"+name[1]));
				context.write(child, new Text("c+"+name[0]+"+"+name[1]));	
			}
		}
	}
	
	public static class Reduce extends Reducer<Text,Text,Text,Text>{
		private static int cnt = 0;
		public void reduce(Text key,Iterable<Text> values,Context context)throws IOException,InterruptedException{
			if(cnt == 0) {
				context.write(new Text("grandchild"),new Text("grandparent"));
				cnt++;
			}
			ArrayList<String> grand_child =new ArrayList<String>();
			ArrayList<String> grand_parent =new ArrayList<String>();
			for(Text val:values) {
				String parent = new String();
				String child = new String();
				String s = val.toString();
				int i = 2;
				while(s.charAt(i) != '+') {
					child = child + s.charAt(i);
					i++;
				}
				i++;
				while(i < s.length()) {
					parent = parent + s.charAt(i);
					i++;
				}
				if(s.charAt(0) == 'c') {
					grand_parent.add(parent);
				}else {
					grand_child.add(child);
				}
			}
			
			for(int i = 0;i < grand_child.size();i++)
				for(int j = 0;j < grand_parent.size();j++) {
					context.write(new Text(grand_child.get(i)), new Text(grand_parent.get(j)));
				}
		}
	}
	public static void main(String[] args)throws Exception {
		// TODO Auto-generated method stub
		 args = new String[] {"hdfs://localhost:9000/ex1/input3","hdfs://localhost:9000/ex1/output3"}; 
		  Configuration conf = new Configuration();
		  conf.set("fs.defaultFS",  "hdfs://localhost:9000");
		  String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		  if (otherArgs.length != 2) {
			  System.err.println("Usage: wordcount <in> <out>");
			  System.exit(2);
		  }
		  Path path =new Path(args[1]);
	      //加载配置文件 
	      FileSystem fileSystem = path.getFileSystem(conf); 
	      //输出目录若存在则删除 
	      if (fileSystem.exists(new Path(args[1]))) { 
	      	fileSystem.delete(new Path(args[1]),true); 
	      }
		  Job job = new Job(conf, "lab3");
		  job.setJarByClass(Lab2.class);
		  job.setMapperClass(Map.class);
		 
		  job.setReducerClass(Reduce.class);
		  job.setOutputKeyClass(Text.class);
		  job.setOutputValueClass(Text.class);
		  FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		  FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		  System.exit(job.waitForCompletion(true) ? 0 : 1);

	}

}
